package com.bigdata.spark.streaming

import java.util.Properties

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}

import scala.collection.mutable.ListBuffer
import scala.util.Random

/**
 * @author Gerry chan
 * @version 1.0
 *  2020/12/29 21：12
 * 案例分析，数据准备, 生产数据并发送到kafka
 * https://www.bilibili.com/video/BV11A411L7CK?p=201
 */
object SparkStreaming10_MockData {
  def main(args: Array[String]): Unit = {
    //生产模拟数据
    //时间戳 省份 城市 用户 广告
    //数据流：Application =>KafKa =>SparkStreaming=>Analysis
    val prop = new Properties()
    //添加配置
    prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.115:9092")
    prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")

    val producer = new KafkaProducer[String, String](prop)

    while (true) {
      mockData().foreach(
        data => {
          // 向Kafka中生成数据
          val record = new ProducerRecord[String, String]("atguiguNew", data)
          producer.send(record)
          println(data)
        }
      )
      Thread.sleep(2000)
    }

    def mockData() = {
      val list = ListBuffer[String]()
      val areaList = ListBuffer[String]("华北","华东","华南")
      val cityList = ListBuffer[String]("北京", "上海", "深圳")

      for (i <- 1 to new Random().nextInt(50)) {
        val area = areaList(new Random().nextInt(3))
        val city = cityList(new Random().nextInt(3))
        var userId = new Random().nextInt(6)+1
        var adid = new Random().nextInt(6) +1

        list.append(s"${System.currentTimeMillis()} ${area} ${city} ${userId}")
      }

      list
    }

  }

}
